package com.rem.flink.flink7Stream;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 基于时间的合流——双流联结（Join）
 * 窗口联结（Window CoGroup）与window join类似  将.join换成.coGroup
 *
 * @author Rem
 * @date 2022-10-21
 */

public class CoGroupTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Long>> stream1 = env.fromElements(
                Tuple2.of("a", 1000L),
                Tuple2.of("b", 1000L),
                Tuple2.of("a", 2000L),
                Tuple2.of("b", 2000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((tuple2, recordTimestamp) -> tuple2.f1));

        SingleOutputStreamOperator<Tuple2<String, Long>> stream2 = env.fromElements(
                Tuple2.of("a", 3000L),
                Tuple2.of("b", 3000L),
                Tuple2.of("a", 4000L),
                Tuple2.of("b", 4000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((tuple2, recordTimestamp) -> tuple2.f1));


        /**
         * 固定搭配
         * JoinFunciton 并不是真正的“窗口函数”，它只是定义了窗口函数在调用时对匹配数据的具体处理逻辑。
         * 将两条流做笛卡尔积 然后处理
         *
         *
         */
        stream1.coGroup(stream2)
                .where(r1 -> r1.f0).equalTo(r2 -> r2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    /**
                     *
                     * 前两个参数不再是单独的每一组“配对”数据了，而是传入了可遍历的数据集合。也就是说，现在不会再
                     * 去计算窗口中两条流数据集的笛卡尔积，而是直接把收集到的所有数据一次性传入至于要怎
                     * 样配对完全是自定义的。这样.coGroup()方法只会被调用一次，而且即使一条流的数据没有任
                     * 何另一条流的数据匹配，也可以出现在集合中、当然也可以定义输出结果了
                     *
                     * coGroup比join更加灵活
                     * 形参:
                     * first -来自第一个输入的记录。
                     * seconds -来自seconds的记录。
                     * out -用于返回元素的收集器。
                     * @throws Exception
                     */
                    @Override
                    public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Long>> second, Collector<String> out) throws Exception {
                        out.collect(first + "=>" + second);
                    }
                })
                .print();

        env.execute();

    }
}
